package hust.rpc.service;

import hust.rpc.annotation.Service;
import hust.rpc.constants.Constants;
import hust.rpc.handler.RegistryHandler;
import hust.rpc.invoker.LocalInvoker;
import hust.rpc.pojo.Host;
import hust.rpc.pojo.Node;
import hust.rpc.utils.CommonUtils;
import hust.rpc.utils.SingletonUtils;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 注册服务实现类
 * 处理提供者节点上下线后信息的维护、获取指定服务集合的提供者节点信息
 */
@Slf4j
@Service(Constants.REGISTRY)
public class RegistryServiceImpl implements ProviderService, ConsumerService {

    // <服务, 提供者集合>映射
    private final Map<String, Set<Host>> serviceToHosts = new ConcurrentHashMap<>();
    // <提供者, 服务集合>映射
    private final Map<Host, List<String>> hostToServices = new ConcurrentHashMap<>();

    // <提供者与注册中心连接的ip端口信息, 提供者提供服务的本地监听ip端口信息>映射
    // 注册中心与提供者的连接不等同于提供者提供服务的本地监听连接
    private final Map<Host, Host> registryHostToBindHost = new ConcurrentHashMap<>();

    /**
     * 注册服务提供者
     *
     * @param host     提供方绑定端口信息
     * @param services 服务接口集合
     */
    @Override
    public void register(Host host, List<String> services) {
        // 获取线程上下文中的Channel(提供者连接注册中心的通道)
        Channel channel = LocalInvoker.getThreadBindChannel();
        Host h = CommonUtils.channelToHost(channel);

        log.info("注册服务提供者{} -> 服务集合: {}", h, services);

        // 更新Host映射
        registryHostToBindHost.put(h, host);

        // 获取单例Handler处理器
        RegistryHandler registryHandler = SingletonUtils.getInstance(RegistryHandler.class);
        // 更新上线服务提供者节点
        registryHandler.updateProviders(new Node(true, System.currentTimeMillis(), host, services));

        // 以下操作为了维护<服务, 提供者节点>映射
        // 先清除原先的映射
        hostToServices.remove(host);
        // 更新映射
        hostToServices.put(host, services);
        // 更新反向映射
        for (String service : services) {
            Set<Host> hosts = serviceToHosts.get(service);
            if (hosts != null) {
                synchronized (hosts) {
                    hosts.add(host);
                }
            } else {
                synchronized (serviceToHosts) {
                    hosts = serviceToHosts.get(service);
                    if (hosts != null) {
                        synchronized (hosts) {
                            hosts.add(host);
                        }
                        continue;
                    }
                    hosts = new HashSet<>();
                    hosts.add(host);
                    serviceToHosts.put(service, hosts);
                }
            }
        }
    }

    /**
     * 取消注册服务提供者(注册中心检测到服务提供者下线后,在单线程的线程池中进行服务提供者的被动下线)
     *
     * @param channel 服务提供者与注册中心的Netty Channel
     */
    public void deRegister(Channel channel) {
        Host bindHost = registryHostToBindHost.get(CommonUtils.channelToHost(channel));
        deRegister(bindHost);
    }

    /**
     * 取消注册服务提供者(服务提供者关闭进程后,触发JVM的Hook钩子时进行远程调用,服务提供者主动下线)
     *
     * @param host 提供方绑定端口信息
     */
    @Override
    public void deRegister(Host host) {
        // 获取线程上下文中的Channel(提供者连接注册中心的通道)
        Channel channel = LocalInvoker.getThreadBindChannel();

        // 清除Host映射
        registryHostToBindHost.remove(CommonUtils.channelToHost(channel));

        // 获取单例Handler处理器
        RegistryHandler registryHandler = SingletonUtils.getInstance(RegistryHandler.class);
        // 更新下线服务提供者节点
        registryHandler.updateProviders(new Node(false, System.currentTimeMillis(), host, null));

        // 以下操作为了维护<服务, 提供者节点>映射
        // 清除正向映射
        List<String> services = hostToServices.remove(host);
        // 清除反向映射
        for (String service : services) {
            Set<Host> hosts = serviceToHosts.get(service);
            synchronized (hosts) {
                hosts.remove(host);
            }
        }
    }

    /**
     * 提供服务消费方感兴趣的服务集合对应的服务提供者信息
     *
     * @param services 服务接口集合
     * @return Map<String, List < Host>> <服务, 服务提供者集合>映射
     */
    @Override
    public Map<String, List<Host>> pullProviders(List<String> services) {
        // 获取线程上下文中的Channel(消费者连接注册中心的通道)
        Channel channel = LocalInvoker.getThreadBindChannel();

        // 获取单例Handler处理器
        RegistryHandler registryHandler = SingletonUtils.getInstance(RegistryHandler.class);
        // 更新上线消费者节点
        registryHandler.updateConsumers(channel, true);

        // 以下操作为了构造请求服务集合对应的服务提供者信息
        Map<String, List<Host>> result = new HashMap<>();
        for (String service : services) {
            Set<Host> hosts = serviceToHosts.get(service);
            if (hosts != null) {
                result.put(service, new ArrayList<>(hosts));
            }
        }
        return result;
    }
}
